package my.data.x.job.thread;

import com.alibaba.datax.core.Engine;
import my.data.x.exception.XErrorCode;
import my.data.x.exception.XException;
import my.data.x.model.Job;
import my.data.x.service.JobService;
import my.data.x.util.JobStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;


import java.util.HashSet;
import java.util.concurrent.*;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

@Service
public class ThreadPool {
    @Autowired
    private JobService jobService;
    public static HashSet<Long> runJobs=new HashSet<>();
    public static Lock lock=new ReentrantLock();
    public static ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1, 1024, 10, TimeUnit.SECONDS, new LinkedBlockingDeque<>(8192));
    public static ExecutorCompletionService<Void> executorCompletionService = new ExecutorCompletionService(threadPoolExecutor);
    public void addThread(Job job){
        executorCompletionService.submit(new JobCallable(job,jobService));
    }
    class JobCallable implements Callable<Void>{
        private Job job;
        private JobService jobService;
        public JobCallable(Job job, JobService jobService){
            this.job=job;
            this.jobService=jobService;
        }
        @Override
        public Void call() throws Exception {
            String config=job.getConfig();

            try{
                lock.lock();
                job=jobService.getJobByid(job.getId());
                if(job.getStatus()!=JobStatus.getStatusTag(JobStatus.RUNNING)&&!runJobs.contains(job.getId())){
                    runJobs.add(job.getId());
                    job.setStatus(JobStatus.getStatusTag(JobStatus.RUNNING));
                    jobService.setJobStatus(job);
                    lock.unlock();
                    Engine.entryDirect(config,job.getId());
                    job.setStatus(JobStatus.getStatusTag(JobStatus.SUCCESS));
                    jobService.setJobStatus(job);
                }else {
                    //直接被cache捕捉到所以不用unlock
                    throw new XException(XErrorCode.RUNTION_ERROR,String.format("作业正在运行，作业状态是 %d",job.getSchedule()));
                }

            }
            catch (Exception e){
                lock.unlock();
                throw e;
            }catch (Throwable throwable){
                lock.unlock();
                job.setStatus(JobStatus.getStatusTag(JobStatus.FAILD));
                jobService.setJobStatus(job);
                throw new XException(XErrorCode.RUNTION_ERROR,String.format("作业执行时报错"),throwable);
            }finally {
                runJobs.remove(job.getId());
            }

            return null;
        }
    }
}
